package job

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/elastic/go-elasticsearch/v8"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/segmentio/kafka-go"
	"review-job/internal/conf"
)

// 自定义执行job结构体，实现transport.Server
type Job struct {
	kafkaReader *kafka.Reader
	es          *ESClient
	log         *log.Helper
}

func NewJob(kafkaReader *kafka.Reader, es *ESClient, logger log.Logger) *Job {
	return &Job{kafkaReader: kafkaReader, es: es, log: log.NewHelper(logger)}
}
func NewKafkaReader(cfg *conf.Kafka) *kafka.Reader {
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers: cfg.Brokers,
		GroupID: cfg.GroupId,
		Topic:   cfg.Topic,
	})
}

type ESClient struct {
	client *elasticsearch.TypedClient
	index  string
}

// 定义kafka中接受到的数据
type Msg struct {
	Type     string                   `json:"type"`
	Database string                   `json:"database"`
	Table    string                   `json:"table"`
	IsDdl    bool                     `json:"isDdl"`
	Data     []map[string]interface{} `json:"data"`
}

// 程序启动干活的
func (j *Job) Start(ctx context.Context) error {
	//评价数据流处理任务
	//1.从kafka中获取MYSQL的数据变更信息
	for {
		m, err := j.kafkaReader.ReadMessage(ctx)
		if errors.Is(err, context.Canceled) {
			return nil
		}
		if err != nil {
			j.log.Errorf("kafka.ReadMessage failed : %v", err)
			break
		}
		data := new(Msg)
		// 2.将数据写入ES
		//实际业务中需要处理数据，例如将几张表的数据联合再插入
		err = json.Unmarshal(m.Value, data)
		if err != nil {
			j.log.Errorf("json.Unmarshal failed : %v", err)
			continue
		}
		if data.Type == "INSERT" {
			for idx := range data.Data {
				j.indexDocument(data.Data[idx])
			}
		} else {
			for idx := range data.Data {
				j.updateDocument(data.Data[idx])
			}
		}
	}
	return nil
}

// 程序结束会调用的
func (j *Job) Stop(ctx context.Context) error {

	return j.kafkaReader.Close()
}

func NewESClient(c *conf.ES) (*ESClient, error) {
	// ES 配置
	cfg := elasticsearch.Config{
		Addresses: c.Addresses,
	}
	// 创建客户端连接
	client, err := elasticsearch.NewTypedClient(cfg)
	if err != nil {
		fmt.Printf("elasticsearch.NewTypedClient failed, err:%v\n", err)
		return nil, err
	}
	return &ESClient{client: client, index: c.Index}, nil
}

func (j *Job) indexDocument(d map[string]interface{}) {
	reviewid := d["review_id"].(string)
	// 添加文档
	resp, err := j.es.client.Index(j.es.index).
		Id(reviewid).
		Document(d).
		Do(context.Background())
	if err != nil {
		j.log.Errorf("indexing document failed, err:%v\n", err)
		return
	}
	j.log.Debugf("result:%#v\n", resp.Result)
}

// updateDocument 更新文档
func (j *Job) updateDocument(d map[string]interface{}) {
	reviewid := d["review_id"].(string)
	resp, err := j.es.client.Update(j.es.index, reviewid).
		Doc(d). // 使用结构体变量更新
		Do(context.Background())
	if err != nil {
		j.log.Errorf("update document failed, err:%v\n", err)
		return
	}
	j.log.Debugf("result:%v\n", resp.Result)
}
